package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.heima.article.mapper.ApArticleConfigMapper;
import com.heima.article.mapper.ApArticleContentMapper;
import com.heima.article.mapper.ApArticleMapper;
import com.heima.model.article.pojos.ApArticle;
import com.heima.model.article.pojos.ApArticleConfig;
import com.heima.model.article.pojos.ApArticleContent;
import com.heima.model.search.pojos.SearchArticleVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.HashMap;

/**
 * @author xiaquanen
 * @version 1.0
 * @project heima-leadnews
 * @date 2024/7/29 10:55:06
 */
@Component
@Slf4j
public class KafkaListen {

    @Autowired
    private ApArticleConfigMapper apArticleConfigMapper;

    @Autowired
    private ApArticleMapper apArticleMapper;

    @Autowired
    private ApArticleContentMapper apArticleContentMapper;

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @KafkaListener(topics = "wm_user_article_xiaqe")
    public void onMessage(String message) {
        HashMap hashMap = JSON.parseObject(message, HashMap.class);
        log.info("消费消息：" + message);
        // 解析消息
        Long articleId = Long.valueOf(hashMap.get("articleId").toString());
        int enable = Integer.parseInt(hashMap.get("enable").toString());
        Integer isDown=enable==0?1:0;
        //同步到索引库中去
        if(isDown==1)
        {
            //索引库中删除数据
            kafkaTemplate.send("topic_delete_es",JSON.toJSONString(articleId));
        }else {
            //索引库中添加数据
            ApArticle apArticle = apArticleMapper.selectById(articleId);
            ApArticleContent apArticleContent = apArticleContentMapper.selectOne(Wrappers.<ApArticleContent>lambdaQuery().eq(ApArticleContent::getArticleId, articleId));
            SearchArticleVo searchArticleVo = new SearchArticleVo();
            BeanUtils.copyProperties(apArticle, searchArticleVo);
            searchArticleVo.setContent(apArticleContent.getContent());
            kafkaTemplate.send("topic_add_es",JSON.toJSONString(searchArticleVo));
        }

        //调用mapper
        apArticleConfigMapper.update(null, Wrappers.<ApArticleConfig>lambdaUpdate().eq(ApArticleConfig::getArticleId,articleId).set(ApArticleConfig::getIsDown,isDown));
        log.info("下架成功");
    }

    @Transactional(rollbackFor = Exception.class)
    @KafkaListener(topics = "topic_deleteArticle")
    public void deleteArticle(String message)
    {
        Long articleId = JSON.parseObject(message, Long.class);
        log.info("删除文章："+articleId);
        apArticleConfigMapper.update(null, Wrappers.<ApArticleConfig>lambdaUpdate().eq(ApArticleConfig::getArticleId,articleId).set(ApArticleConfig::getIsDown,1));
    }
}
